use std::collections::HashMap;

use model_graph_types::{
    connector::{Connector, ExecuteOptions, QueryOptions},
    database::DataSource,
    modeling::Value,
};

use dgraph_tonic::{Client, Endpoint, EndpointConfig, Mutate, Mutation, Query, Response};
use std::time::Duration;

pub struct DgraphConnector {
    client: Client,
}

#[derive(Debug, Default)]
struct EndpointWithTimeout {}

impl EndpointConfig for EndpointWithTimeout {
    fn configure_endpoint(&self, endpoint: Endpoint) -> Endpoint {
        endpoint.timeout(Duration::from_secs(5))
    }
}

impl DgraphConnector {
    pub async fn open(data_source: &DataSource) -> anyhow::Result<DgraphConnector> {
        let endpoint_config = EndpointWithTimeout::default();

        let client = Client::new_with_endpoint_config(&data_source.url, endpoint_config)?;
        let connector = DgraphConnector { client };
        Ok(connector)
    }
}

#[async_trait::async_trait]
impl Connector for DgraphConnector {
    async fn query(
        &self,
        sql: &String,
        args: &Vec<Value>,
        options: QueryOptions,
    ) -> anyhow::Result<(Vec<Value>, Vec<Value>)> {
        tracing::debug!("开始查询......");

        let mut txn = self.client.new_read_only_txn();

        let rows = if args.len() > 0 {
            //添加参数
            let mut vars = HashMap::new();
            for (i, arg) in args.iter().enumerate() {
                vars.insert(
                    format!("${}", i + 1),
                    TryInto::<String>::try_into(arg.clone())
                        .map_err(|err| anyhow::anyhow!("参数不能转换为字符串!"))?,
                );
            }
            let response: Response = txn.query_with_vars(sql, vars).await?;
            response
        } else {
            let response: Response = txn.query(sql).await?;
            response
        };
        tracing::debug!("进行值的类型转换!");
        let values: serde_json::Value = rows.try_into()?;

        match values {
            serde_json::Value::Array(vs) => Ok((
                vec![],
                vs.iter()
                    .map(|sv| Value::try_from(sv).map_or_else(|_| Value::None, |v| v))
                    .collect::<Vec<Value>>(),
            )),
            _ => Ok((vec![], vec![Value::try_from(values)?])),
        }
    }

    async fn query_one(
        &self,
        sql: &String,
        args: &Vec<Value>,
        options: QueryOptions,
    ) -> anyhow::Result<(Vec<Value>, Value)> {
        tracing::debug!("开始查询......");

        let mut txn = self.client.new_read_only_txn();

        let rows = if args.len() > 0 {
            //添加参数
            let mut vars = HashMap::new();
            for (i, arg) in args.iter().enumerate() {
                vars.insert(
                    format!("${}", i + 1),
                    TryInto::<String>::try_into(arg.clone())
                        .map_err(|err| anyhow::anyhow!("参数不能转换为字符串!"))?,
                );
            }
            let response: Response = txn.query_with_vars(sql, vars).await?;
            response
        } else {
            let response: Response = txn.query(sql).await?;
            response
        };
        tracing::debug!("进行值的类型转换!");
        let value: serde_json::Value = rows.try_into()?;

        Ok((vec![], value.try_into()?))
    }

    async fn query_map(
        &self,
        sql: &String,
        args: &Vec<Value>,
        options: QueryOptions,
    ) -> anyhow::Result<HashMap<String, Value>> {
        if let (headers, Value::Array(data)) = self.query_one(sql, args, options).await? {
            Ok(_to_object(&headers, data)?)
        } else {
            Err(anyhow::anyhow!("查询失败"))
        }
    }

    async fn query_maps(
        &self,
        sql: &String,
        args: &Vec<Value>,
        options: QueryOptions,
    ) -> anyhow::Result<Vec<HashMap<String, Value>>> {
        if let (headers, data) = self.query(sql, args, options).await? {
            let mut res = vec![];
            for row in data {
                match row {
                    Value::Array(v) => res.push(_to_object(&headers, v)?),
                    _ => {}
                }
            }
            Ok(res)
        } else {
            Err(anyhow::anyhow!("查询失败"))
        }
    }

    async fn execute(
        &self,
        sql: &String,
        args: &Vec<Value>,
        options: ExecuteOptions,
    ) -> anyhow::Result<i64> {
        tracing::debug!("开始执行......");

        let mutate_upsert = _get_value_from_option(&options.options, "type", "mutate")?;

        let mut txn = self.client.new_mutated_txn();

        //
        let rows = if mutate_upsert == "upsert" {
            let mutation_sql = _get_value_from_option(&options.options, "mutation", "")?;
            let mut mu = Mutation::new();
            mu.set_set_json(&mutation_sql)
                .map_err(|err| anyhow::anyhow!("{}", err))?;

            //
            if args.len() > 0 {
                //添加参数
                let mut vars = HashMap::new();
                for (i, arg) in args.iter().enumerate() {
                    vars.insert(
                        format!("${}", i + 1),
                        TryInto::<String>::try_into(arg.clone())
                            .map_err(|err| anyhow::anyhow!("参数不能转换为字符串!"))?,
                    );
                }
                let response: Response = txn.upsert_with_vars(sql, vars, mu).await?;
                response
            } else {
                let response: Response = txn.upsert(sql, mu).await?;
                response
            }
        } else {
            let mut mu = Mutation::new();
            mu.set_set_json(&sql)
                .map_err(|err| anyhow::anyhow!("{}", err))?;
            //
            let response: Response = txn.mutate(mu).await?;
            response
        };

        tracing::debug!("进行值的类型转换!");
        let value: serde_json::Value = rows.try_into()?;
        tracing::debug!("执行完成......{}", value);

        txn.commit().await.expect("Transaction is commited");

        // if res.last_insert_id() > 0 {
        //     Ok(res.last_insert_id() as i64)
        // } else {
        //     Ok(i64::try_from(res.rows_affected())?)
        // }
        // Err(anyhow::anyhow!(""))
        Ok(1)
    }

    async fn close(&mut self) -> anyhow::Result<i32> {
        Ok(0)
    }
}

fn _get_value_from_option(
    options: &HashMap<String, String>,
    key_str: &str,
    default_value: &str,
) -> anyhow::Result<String> {
    let key = String::from(key_str);

    if options.contains_key(&key) {
        if let Some(v) = options.get(&key) {
            return Ok(v.clone());
        }
    }

    Ok(String::from(default_value))
}

fn _to_object(headers: &Vec<Value>, data: Vec<Value>) -> anyhow::Result<HashMap<String, Value>> {
    let mut object = HashMap::new();

    for i in 0..headers.len() {
        if let Some(header) = headers.get(i) {
            match header {
                Value::Object(hm) => {
                    if let Some(name) = hm.get(&String::from("name")) {
                        match name {
                            Value::String(hv) => {
                                if let Some(value) = data.get(i) {
                                    object.insert(hv.clone(), value.clone());
                                }
                            }
                            _ => {}
                        }
                    }
                }
                Value::String(hv) => {
                    if let Some(value) = data.get(i) {
                        object.insert(hv.clone(), value.clone());
                    }
                }
                _ => {}
            }
        }
    }
    Ok(object)
}
